/*
 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 */

/*
 *
 *
 *
 *
 *
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */

package java.util.concurrent;
import java.util.*;

/**
 * Provides default implementations of {@link ExecutorService}
 * execution methods. This class implements the {@code submit},
 * {@code invokeAny} and {@code invokeAll} methods using a
 * {@link RunnableFuture} returned by {@code newTaskFor}, which defaults
 * to the {@link FutureTask} class provided in this package.  For example,
 * the implementation of {@code submit(Runnable)} creates an
 * associated {@code RunnableFuture} that is executed and
 * returned. Subclasses may override the {@code newTaskFor} methods
 * to return {@code RunnableFuture} implementations other than
 * {@code FutureTask}.
 *
 * <p><b>Extension example</b>. Here is a sketch of a class
 * that customizes {@link ThreadPoolExecutor} to use
 * a {@code CustomTask} class instead of the default {@code FutureTask}:
 *  <pre> {@code
 * public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
 *
 *   static class CustomTask<V> implements RunnableFuture<V> {...}
 *
 *   protected <V> RunnableFuture<V> newTaskFor(Callable<V> c) {
 *       return new CustomTask<V>(c);
 *   }
 *   protected <V> RunnableFuture<V> newTaskFor(Runnable r, V v) {
 *       return new CustomTask<V>(r, v);
 *   }
 *   // ... add constructors, etc.
 * }}</pre>
 *
 * @since 1.5
 * @author Doug Lea
 */
public abstract class AbstractExecutorService implements ExecutorService {

    /**
     * Returns a {@code RunnableFuture} for the given runnable and default
     * value.
     *
     * @param runnable the runnable task being wrapped
     * @param value the default value for the returned future
     * @param <T> the type of the given value
     * @return a {@code RunnableFuture} which, when run, will run the
     * underlying runnable and which, as a {@code Future}, will yield
     * the given value as its result and provide for cancellation of
     * the underlying task
     * @since 1.6
     */
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new FutureTask<T>(runnable, value);
    }

    /**
     * Returns a {@code RunnableFuture} for the given callable task.
     *
     * @param callable the callable task being wrapped
     * @param <T> the type of the callable's result
     * @return a {@code RunnableFuture} which, when run, will call the
     * underlying callable and which, as a {@code Future}, will yield
     * the callable's result as its result and provide for
     * cancellation of the underlying task
     * @since 1.6
     */
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        // 1. new 一个 FutureTask，将任务包装成 FutureTsk
        RunnableFuture<T> ftask = newTaskFor(task);
        // 2. 交给执行器执行，execute 方法由具体的子类来实现
        execute(ftask);
        return ftask;
    }

    /**
     * the main mechanics of invokeAny.
     * <pre>
     * 1. 此方法的目的
     *      将 tasks 集合中的任务 提交到线程池执行，只要任意一个任务执行完后就可以结束了。
     *      只获取【第一个】成功完成任务的执行结果。一旦有任务完成，其他的任务就会通过 Future.cancel(true)方法 直接取消。
     * 2. 实际效果
     *      除了第一个成功执行完的任务，其他的任务都会通过 Future.cancel()方法取消，
     *    而 Future.cancel(true)方法则会调用 执行任务的线程的interrupt()中断方法。
     * 1). 如果 其他任务 响应中断，即，在任务代码中触发 InterruptedException异常 或 主动检测到中断标志，那么，其他任务就会被中断（Thread.interrupt()），从而退出执行。
     * 2). 如果 其他任务 不响应中断，那么，即使调用了 Future.cancel()方法，任务也会继续执行，直到自然结束。
     * </pre>
     * @param tasks 提交的任务（待执行的任务）集合
     */
    private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                              boolean timed, long nanos)
        throws InterruptedException, ExecutionException, TimeoutException {
        if (tasks == null)
            throw new NullPointerException();
        // 待执行任务的数量
        int ntasks = tasks.size();
        if (ntasks == 0)
            throw new IllegalArgumentException();
        //
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
        /**
         * 核心实现就依靠这个东西了。详情就直接参见 {@link ExecutorCompletionService}，
         * 这个东西的核心就在于它会把 异步任务执行的结果 放到一个队列中，哪个异步任务先执行完就先放入队列中，
         * 所以说，我们只要拿到队列中的第一个值（即，第一个执行完成的异步任务）就算结束了，其他的异步任务就直接取消。
         *
         * {@link ExecutorCompletionService#completionQueue}
         */
        ExecutorCompletionService<T> ecs =
            new ExecutorCompletionService<T>(this);

        // For efficiency, especially in executors with limited
        // parallelism, check to see if previously submitted tasks are
        // done before submitting more of them. This interleaving
        // plus the exception mechanics account for messiness of main
        // loop.

        try {
            // Record exceptions so that if we fail to obtain any
            // result, we can throw the last exception we got. 记录可能抛出的异常
            ExecutionException ee = null;
            final long deadline = timed ? System.nanoTime() + nanos : 0L;// 初始化超时时间
            Iterator<? extends Callable<T>> it = tasks.iterator();

            // Start one task for sure; the rest incrementally
            // 先提交一个任务，其他的任务在下面的 for 自旋中一个个提交
            futures.add(ecs.submit(it.next()));
            // 提交了一个任务，所以任务数量减 1
            --ntasks;
            // 正在执行的任务数量（提交的时候 +1，任务执行结束的时候 -1）
            int active = 1;

            // 自旋，直到拿到其中一个异步任务的执行结果。
            for (;;) {
                // esc（ExecutorCompletionService） 内部有一个 completionQueue 队列用于保存执行完成的结果。
                // ecs.poll() 本质上调用的还是 completionQueue.poll()方法。
                // BlockingQueue#poll() 方法不阻塞，返回为 null 表示队列为空
                Future<T> f = ecs.poll();
                /**
                 * f 为 null，说明在 for(;;) 自旋之前提交的第一个任务还没有执行完成。
                 * 当然，也说明当前还没有任何一个任务执行完成。
                 * 在前面提交了一个任务，然后在这里做一次检查，也是为了提高性能。
                 */
                if (f == null) {
                    // 如果 ntasks 大于 0，那就说明还存在没有提交的任务。
                    if (ntasks > 0) {
                        --ntasks;
                        // 提交任务
                        futures.add(ecs.submit(it.next()));
                        ++active;
                    }
                    /**
                     * ① 代码走到这里，说明 ntasks == 0 为真（即，当前没有可提交的任务了）。
                     * ② 若 active == 0 也为真，则说明所有任务都执行失败。
                     *    也就是说，所有线程在走到下面 if(f != null) 的代码分支内部都执行失败了。
                     */
                    else if (active == 0)
                        // 所有任务都执行失败了，那就 break，结束 for(;;)自旋。
                        break;
                    else if (timed) {
                        f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                        if (f == null)
                            throw new TimeoutException();
                        nanos = deadline - System.nanoTime();
                    }
                    /**
                     * 代码执行到这里，则说明：
                     *      ① ntasks == 0 为真，即，当前没有可提交的任务了；
                     *      ② active != 0 为真，即，当前还存在没有执行完成的线程；
                     *      ③ timed 为 false，即，没有设置超时。
                     */
                    else
                        // take 方法会阻塞，直到有元素返回。
                        f = ecs.take();
                }
                // future 不为空，就代表已经有一个异步任务已经执行完成。
                if (f != null) {
                    // 如果 future 不为空，则 active（正在执行的任务数量）减一
                    --active;
                    try {
                        /**
                         * 获取 {@link ExecutorCompletionService#completionQueue} 队列中 Future 的执行结果，并结束 for 自旋。
                         */
                        return f.get();
                    } catch (ExecutionException eex) {
                        ee = eex;
                    } catch (RuntimeException rex) {
                        ee = new ExecutionException(rex);
                    }
                }
            }// 注意这里，for(;;) 自旋的结束范围

            /**
             * 1. 代码执行到这里的原因：【只有一种】
             *    就是上面的 for(;;) 自旋结束了。
             *    那么 for(;;) 自旋结束，且还能执行到下面这些代码的方案只有一种，那就是 for(;;) 内部 break 了，
             *    因为 return 或 throw exception 后，代码是执行不到这里的，只会执行 finally {} 代码块了。
             * 2. 再来看上面 for(;;) 自旋 break 的情况，通过上面的分析可知，只有当所有任务都执行失败时，才会 break。
             */
            if (ee == null)
                ee = new ExecutionException();
            throw ee;

        } finally {
            // for 循环取消正在执行的其他异步任务，因为我们在 try 代码块的 for 自旋中已经拿到了第一个执行完成的异步任务的结果，
            // 所以，其他的异步任务就没有继续执行下去的必要了。
            for (int i = 0, size = futures.size(); i < size; i++)
                // 调用 FutureTask#cancel 方法，入参为 true，表示直接中断线程。
                futures.get(i).cancel(true);
        }
    }

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
        try {
            return doInvokeAny(tasks, false, 0);
        } catch (TimeoutException cannotHappen) {
            assert false;
            return null;
        }
    }

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                           long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        return doInvokeAny(tasks, true, unit.toNanos(timeout));
    }

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        // 标识所有任务是否完成
        boolean done = false;
        try {
            for (Callable<T> t : tasks) {
                // 将任务包装成 FutureTsk
                RunnableFuture<T> f = newTaskFor(t);
                futures.add(f);
                // 交给执行器执行
                execute(f);
            }

            // for 循环获取所有线程的执行结果
            for (int i = 0, size = futures.size(); i < size; i++) {
                Future<T> f = futures.get(i);
                if (!f.isDone()) {
                    try {
                        // 阻塞等待任务完成并返回结果
                        f.get();
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    }
                }
            }
            done = true;
            // ① 这个方法返回的不像其他的场景，返回一个 List<Future<T>>，其实执行结果还没有出来。
            // ② 这个方法就不一样了，因为在上面的 for 循环中已经逐个去获取异步任务的执行结果了，
            //    所以说，到这里所有任务其实都已经执行完成了。
            return futures;
        } finally {
            /**
             * Q: 什么情况下 done 为 false 呢（也就是 try 代码块中【done = true;】这一行代码执行不到）？
             *    肯定只能是出现异常了。那又是什么异常呢？
             * A: 上面的【f.get();】这一步可能会抛出 InterruptedException（中断异常）。
             */
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                         long timeout, TimeUnit unit)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Callable<T> t : tasks)
                futures.add(newTaskFor(t));

            final long deadline = System.nanoTime() + nanos;
            final int size = futures.size();

            // Interleave time checks and calls to execute in case
            // executor doesn't have any/much parallelism.
            for (int i = 0; i < size; i++) {
                execute((Runnable)futures.get(i));
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L)
                    return futures;
            }

            for (int i = 0; i < size; i++) {
                Future<T> f = futures.get(i);
                if (!f.isDone()) {
                    if (nanos <= 0L)
                        return futures;
                    try {
                        f.get(nanos, TimeUnit.NANOSECONDS);
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    } catch (TimeoutException toe) {
                        return futures;
                    }
                    nanos = deadline - System.nanoTime();
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }

}
